package org.liu.reactor;

import java.util.concurrent.Flow;
import java.util.concurrent.SubmissionPublisher;

public class FlowDemo {

    public static void main(String[] args) throws InterruptedException {

        SubmissionPublisher<String> publisher = new SubmissionPublisher<String>();


        Flow.Subscriber<String> subscriber = new Flow.Subscriber<>() {
            Flow.Subscription subscription;

            @Override
            public void onSubscribe(Flow.Subscription subscription) {
                this.subscription = subscription;
                subscription.request(1);
            }

            @Override
            public void onNext(String item) {
                System.out.println("收到数据："+item);
                subscription.request(1);
            }

            @Override
            public void onError(Throwable throwable) {
                System.out.println("报错了"+throwable.getMessage());
            }

            @Override
            public void onComplete() {
                System.out.println("完成了");

            }
        };


        publisher.subscribe(subscriber);

        for (int i = 0; i < 20; i++) {
            publisher.submit("Hello " + i);
            if (i>18){
                publisher.closeExceptionally(new RuntimeException(i+ "异常了"));
            }
        }


        Thread.sleep(10000);
    }
}
